一：DeepAgents 框架定位与核心价值

1.1. 什么是 DeepAgents

DeepAgents 是一个基于 LangChain 和 LangGraph 构建的企业级高级智能体框架。它建立在 LangGraph（底层运行时）和 LangChain（工具 / 模型层）之上，是一个高阶的 Agent Harness（智能体装备 / 套件）**。

DeepAgents 开源地址：https://github.com/langchain-ai/deepagents

1.2. 解决的核心问题

传统的 Agent 开发通常运行一个简单的循环：思考 → 调用工具 → 观察 → 重复。这种模式在处理多小时或多天的任务时，容易遇到以下"浅层陷阱"（Shallow Agent Problem）

1. 规划能力缺失：原生 Agent 倾向于“走一步看一步”，缺乏全局视角的任务拆解，容易在多步任务中迷失方向。
2. 遗忘与混乱：在执行超过 10-20 步的长任务时，由于 Context Window（上下文窗口）限制，传统 Agent 容易忘记初始目标或陷入死循环。
3. 环境交互困难：文件系统操作、代码执行环境（沙箱）的配置和安全管理复杂。
4. 上下文污染：所有工具返回结果都堆积在一个对话历史中，导致噪声过大。
5. 协作编排复杂：多智能体（Multi-Agent）之间的任务分发和上下文隔离难以实现。

DeepAgents 通过引入"类人"的工作流解决了这些问题：先做计划（Plan），再执行，利用文件系统管理记忆，遇到复杂子任务时"外包"给子 Agent。将规划工具、文件系统访问、子代理和详细提示词等关键机制整合在一起，以支持复杂的深度任务 。

1.3. 应用场景

DeepAgents 不适合用来做简单的聊天机器人（Chatbot），它是为重任务设计的，它适用于任务需规划、上下文海量、需多专家协作、要求持久记忆的场景，将 LangChain 生态从单步响应提升至自主完成复杂项目的高度：

• 深度调研 (Deep Research)：自动进行多轮网页搜索、阅读文档、整理笔记并生成长篇报告（如分析某个行业的市场格局）。
• 全栈代码生成 (Coding Assistant)：类似 Claude Code，在沙箱环境中编写、运行、测试和修复代码，甚至重构整个代码库。
• 复杂数据分析：自动连接数据库，生成 SQL，执行查询，将中间数据存为 CSV 文件（在虚拟文件系统中），最后生成图表。
• 自动化运维 (DevOps Automation)：操作文件系统、执行 Shell 命令、管理服务器状态。
• 复杂工作流编排：需要多角色协作（如产品经理-程序员-测试员）的复杂业务流程。

DeepAgents 适用场景

二：与 LangChain 及 LangGraph 的区别

• LangChain：提供 Prompt, Models, Tools 等积木。
• LangGraph：提供 State, Nodes, Edges 等地基和连接逻辑。
• DeepAgents：LangGraph 的一种“最佳实践实现”。它底层使用 LangGraph 来管理状态和循环，但向上提供了更高级的 API (create_deep_agent)，隐藏了底层的图构建细节。

三：DeepAgents 核心功能介绍

3.1. 核心入口：create_deep_agent()

这是整个框架的核心函数，它创建了一个功能完整的深度智能体。

默认配置：

• 使用 Claude Sonnet 4 或 GPT-4o 作为默认模型（推荐）。
• 集成 7 个核心文件操作工具。
• 提供待办事项管理功能。
• 支持子代理调用。

关键参数：

• model: 支持自定义语言模型。
• tools: 自定义工具集。
• system_prompt : 系统提示词
• subagents: 子代理配置。
• backend: 文件存储后端。
• interrupt_on: 人机交互配置 (Human-in-the-Loop)。允许在特定节点暂停 Agent 执行，等待人工干预。这对于安全审核（删除文件）、成本控制（调用昂贵 API）和质量保证至关重要。

from deepagents import create_deep_agent
from langchain_tavily import TavilySearch
from langgraph.checkpoint.memory import InMemorySaver

# 1. 初始化 Tavily 搜索工具
tavily = TavilySearch(max_results=3)

# 2. 编写系统提示词
research_instructions = """
您是一位资深的研究人员。您的工作是进行深入的研究，然后撰写一份精美的报告。
您可以通过互联网搜索引擎作为主要的信息收集工具。
## 可用工具
### `互联网搜索`
使用此功能针对给定的查询进行互联网搜索。您可以指定要返回的最大结果数量、主题以及是否包含原始内容。
### `写入本地文件`
使用此功能将研究报告保存到本地文件。当您完成研究并生成报告后，请使用此工具将完整的报告内容保存到文
件中。
- 文件路径建议使用 .md 格式（Markdown），例如 "research_report.md" 或 "./reports/报告名
称.md"
- 请确保报告内容完整、结构清晰，包含所有章节和引用来源
## 工作流程
在进行研究时：
1. 首先将研究任务分解为清晰的步骤
2. 使用互联网搜索来收集全面的信息
3. 将信息整合成一份结构清晰的报告
4. **重要**：完成报告后，务必使用 `写入本地文件` 工具将完整报告保存到本地文件
5. 务必引用你的资料来源
**注意**：请确保在完成研究后，将完整的报告内容保存到文件中，这样用户可以方便地查看和保存报告。
"""

# 2. 创建 DeepAgents 智能体
agent = create_deep_agent(
    name="DeepAgents_Agent",       # 智能体名称
    tools=[tavily],                # 可调用工具 Tool
    model=model,                   # 模型 Model
    system_prompt=research_instructions,  # 系统提示词
    checkpointer=InMemorySaver(),  # 检查点 Checkpointer，内存检查点
)

# 3. 配置线程 ID
config = {"configurable": {"thread_id": "1"}}

result = agent.invoke({"messages": [{"role": "user", "content": "帮我查询一下有关deepagents框架的最新动态"}]}, config=config)

3.2. create_deep_agent 内部结构

源码参数截图

除了自己定义的工具（如 Tavily 搜索），DeepAgents 还默认添加了一些其他工具：

• 文件系统中间件（FileSystemMiddleware）: 用于读写、查询、执行文件系统中的文件。
• 待办事项中间件（TodoListMiddleware）: write_todos 用于写入待办事项，task 用于创建子 agent 来执行待办事项。

这些都是 DeepAgents 特有的功能，用于支持智能体在实际应用中的各种场景。那么，接下来我们来看看这些功能的具体应用。

• Langgraph Studio 中可视化结构图：能看到
  • PatchToolCallsMiddleware 用于自动检测并修复“悬空”的工具调用的关键中间件，确保工具调用的完整性和正确性。
  • SummarizationMiddleware 上下文压缩中间件，防止上下文过长

四：四大核心内置工具与组件详解

DeepAgents 通过中间件 (Middleware) 的形式，为智能体注入了四项核心能力，构成了框架的四大支柱（Four Pillars）：

其核心价值在于将原本需要手动编排的规划 - 存储 - 委托 - 执行流程，固化为中心化、可复用、可观测的中间件体系，标志着 AI Agent 从"脚本化"向"产品化"的关键演进。

4.1. 系统提示词 (System Prompts)

• 功能：定义 Agent 的“人设”、行为准则和核心目标。
• 机制：框架会自动将用户定义的 system_prompt 与内置的 BASE_AGENT_PROMPT 结合。
• 作用：确保 Agent 始终遵循指令，理解其可用的工具集，并保持一致的输出风格。
• 角色本质：系统提示词是 DeepAgents 三大中间件协同的"契约"，其默认版本包含：
  • 规划指令：要求 LLM 在任务开始前必须调用 write_todos，输出格式为 JSON Schema
  • 文件操作规范：明确 write_file 用于新内容，edit_file 用于局部修改，避免覆盖冲突
  • 子代理调用协议：规定 task 工具的参数结构及结果通过 /subagent_results/.md 回传
  • 安全底线：禁止直接执行删除、格式化等危险命令，必须通过 execute 沙箱

4.2. 规划工具 (Planning System / Todo List)

• 组件：TodoListMiddleware
• 工具名：write_todos
• 功能：Agent 在行动前先生成 Markdown 格式的待办事项列表 (Todo List)，并在执行过程中更新状态（完成 / 进行中）。
• 工作流：
  1. Agent 接收复杂任务（简单短期的任务不会触发 todolist）。
  2. 调用 write_todos 将任务拆解为子步骤 (Pending)。
  3. 每完成一步，更新状态为 (Completed)。
  4. 自我反思：在每一步行动前，Agent 都会看到当前的 Todo List，从而避免迷失方向。这强制模型进行"思维链"的显性化管理。

4.3. 子代理 (Sub-Agent Delegation)

• 组件：SubAgentMiddleware
• 工具名：task (delegate_task)
• 核心概念：
  • 任务隔离：每个子代理有独立的上下文窗口。
  • 并行执行：支持同时启动多个子代理。
  • 结果聚合：智能整合多个子代理的输出。
• 机制：
  • 当任务过于具体（如"爬取并分析这篇长论文"）时，主 Agent 会生成一个隔离环境的子 Agent 去执行。
  • 子 Agent 启动时，只继承必要的环境配置，但拥有全新的、空白的消息历史。
  • 子 Agent 执行完毕后只返回一个总结性的结果。这保证了主 Agent 的时间线（Context）保持干净，极大地节省了 Token。

1. 显示传入 subAgent 参数

pip install langchain-mcp-adapters mcp

import asyncio
import os
from dotenv import load_dotenv
from rich.console import Console
from rich.panel import Panel
from rich.tree import Tree
from deepagents import create_deep_agent
from langchain_deepseek import ChatDeepSeek
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_community.tools import TavilySearchResults

from langchain_core.messages import ToolMessage, BaseMessage

load_dotenv(override=True)

async def setup_mcp_tools():
    try:
        client = MultiServerMCPClient({
            "context7": {
                "transport": "stdio",
                "command": "npx",
                "args": ["-y", "@upstash/context7-mcp@latest"],
            }
        })
        # 获取工具
        tools = await client.get_tools()
        return client, tools
    except Exception as e:
        return None, []

# 定义子 Agent 配置
def get_subagents_config(mcp_tools):
    # 子 Agent 1: 官方文档专家
    doc_tools = mcp_tools if mcp_tools else [TavilySearchResults(max_results=3)]

    docs_researcher = {
        "name": "DocsResearcher",
        "description": "负责查阅官方文档和技术规范的专家 Agent。",
        "system_prompt": "你是一名专门查阅官方文档的技术专家。请使用工具获取准确的技术细节。不要猜测。",
        "tools": doc_tools,
        "model": "deepseek-chat"
    }

    # 子 Agent 2: 社区生态专家
    community_researcher = {
        "name": "CommunityResearcher",
        "description": "负责搜索社区博客、教程和最佳实践的专家 Agent。",
        "system_prompt": "你是一名关注社区动态的开发者。请搜索博客、论坛和 GitHub 讨论。",
        "tools": [TavilySearchResults(max_results=3)],
        "model": "deepseek-chat"
    }

    return [docs_researcher, community_researcher]

# 主运行逻辑
async def run_parallel_demo():
    # 初始化 MCP
    mcp_client, mcp_tools = await setup_mcp_tools()

    # 获取子 Agent 配置
    subagents = get_subagents_config(mcp_tools)

    # 创建主 Agent
    llm = ChatDeepSeek(model="deepseek-chat", temperature=0)

    agent = create_deep_agent(
        model=llm,
        tools=[],
        subagents=subagents,        # 传入子 Agent 配置
        system_prompt="""你是一名技术总监。你的任务是协调 DocsResearcher 和 CommunityResearcher 完成调研任务。
                        请根据用户需求，将任务拆解并分发给这两个子 Agent。
                        如果任务允许，请务必并行调用它们以提高效率。
                        最后汇总它们的报告。"""
    )

    task = "请详细调研 'LangChain DeepAgents' 框架。我需要官方的技术架构说明（来自文档）以及社区的最佳实践案例。请对比两者。"

五：文件系统集成 (Filesystem & Sandbox)

• 组件：FilesystemMiddleware & Backend
• 功能：这是 DeepAgent 的核心"外挂大脑"。Agent 不再将所有检索到的长文塞入上下文（Context），而是将其写入 Filesystem（如 /workspace），仅在需要时读取。这极大地扩展了 Agent 的"工作记忆"。
• 核心工具集：
  • ls: 浏览目录结构。
  • read_file: 读取文件内容。关键封装：支持 offset（偏移量）和 limit（行数），强制 Agent 对大文件进行分页读取，避免一次性读爆上下文。
  • write_file: 创建新文件。
  • edit_file: 修改文件。支持精确的字符串替换和 replace_all 模式，并有防呆设计（要求先读后改）。
  • glob: 通配符模糊查找文件。
  • grep: 正则表达式搜索文件内容，像命令行一样在代码库中定位目标。
  • execute: 执行 Shell 命令（需 Sandbox 支持），用于运行代码、安装依赖等。
• 亮点功能：大结果自动转存
  • 当 Agent 调用工具产生的结果过长时，系统会自动拦截，将完整结果写入文件系统（如 /large_tool_results/{id}），并只给 Agent 返回一个摘要和文件路径。这完美解决了搜索结果或日志文件过长导致 Agent 崩溃的问题。
• 后端支持 (Backends)：
  • FilesystemBackend: 直接操作本地磁盘。
  • DockerBackend: 在 Docker 容器中执行，提供隔离环境。
  • StoreBackend: 存储在数据库中，支持查询和检索。
  • E2BBackend: 使用 E2B 云端沙箱。
  • CompositeBackend: 混合模式（如：本地存文件，Docker 跑代码）。

5.1. 核心工具集

# 使用本地文件系统后端，根目录设为 ./workspace
# 这样我们可以看到真实文件的创建
# 注意：设置 virtual_mode=True 以支持绝对路径 (如 /hello_world.py) 映射到 ./workspace
backend = FilesystemBackend(root_dir="./workspace", virtual_mode=True)

# 创建 Agent
# 注意：create_deep_agent 默认会自动包含 FilesystemMiddleware
agent = create_deep_agent(
    model=model,
    backend=backend,
    system_prompt="你是一个文件系统操作助手。请根据用户指令使用相应的工具。"
)

5.2. Backend 后端应用

1. 默认模式（内存沙箱）

内存沙箱，支持在内存中执行代码，防止代码注入攻击。代码执行结果会被存储在内存中，不会对本地环境造成影响。执行结束后，内存中的数据会被清除。

from deepagents import create_deep_agent
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv(override=True)

model = ChatOpenAI(model="gpt-4o", temperature=0)

# 默认情况下，DeepAgent 会自动加载 FilesystemMiddleware 并使用内存后端
agent = create_deep_agent(
    model=model,
    system_prompt="你是一个数据处理助手。"
)

# Agent 可以自由创建文件、读取文件，但这些文件只存在于内存中
# 任务结束后，这些文件会自动消失
agent.invoke({"messages": [("user", "创建一个名为 test.txt 的文件并写入 'Hello, World!'")]})

2. 持久化模式（操作真实文件）

持久化沙箱，支持将代码执行结果持久化到本地磁盘，防止代码执行过程中数据丢失。代码执行结束后，内存中的数据会被清除，但是磁盘上的文件会被保留。

from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o", temperature=0)

# 将 Agent 的根目录映射到本地的 "./workspace" 文件夹，virtual_mode=True 表示启用虚拟文件系统，作用是将 Agent 执行的所有文件操作都映射到本地的 "./workspace" 文件夹，而不是直接操作本地文件系统。
backend = FilesystemBackend(root_dir="./workspace", virtual_mode=True)

agent = create_deep_agent(
    model=model,
    backend=backend  # 传入后端，中间件会自动使用它
)

# 此时 Agent 执行 write_file("/readme.md", ...) 会在本地 ./workspace/readme.md 创建文件
agent.invoke({"messages": [("user", "创建一个名为 readme.md 的文件并写入 'Hello, World!'")]})

3. 演示大文件读取分页

在 FilesystemBackend 分页的参数是固定死的，默认每次读取 500 行，我们可以在系统提示词中指定每次读取的行数。

import asyncio
import os
import shutil
from pathlib import Path
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, ToolMessage
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend

load_dotenv(override=True)

# 定义工作目录
WORK_DIR = Path("workspace/pagination_demo").resolve()
LARGE_FILE_NAME = "server_logs.txt"
LARGE_FILE_PATH = WORK_DIR / LARGE_FILE_NAME
TARGET_SECRET = "CRITICAL_ERROR_CODE_998877"

async def run_pagination_demo():
    # 初始化 FilesystemBackend
    # 将 backend 指向我们的测试目录
    # virtual_mode=True 确保 Agent 只能访问该目录下的文件，不能访问宿主机其他目录
    backend = FilesystemBackend(root_dir=WORK_DIR, virtual_mode=True)

    # 创建 Agent
    # 我们明确指示 Agent 使用分页读取，每次读取 300 行，Agent 默认每次读取 500 行(DEFAULT_READ_LIMIT)
    system_prompt = """
    你是一个专业的系统管理员。
    你的任务是从日志文件中查找特定的错误代码。
    注意：日志文件可能非常大，为了避免上下文溢出，你必须使用 `read_file` 工具的分页功能。
    每次读取请限制在 300 行以内 (limit=300)，并使用 offset 参数向后滚动。
    直到找到目标信息为止。
    """

    llm = ChatOpenAI(model="gpt-4o", temperature=0)

    agent = create_deep_agent(
        model=llm,
        backend=backend,
        system_prompt=system_prompt
    )

4. E2BBackend 使用 E2B 云沙箱

E2B(Environment To Be) 是一个专为 AI 智能体设计的云端安全执行环境 。当 AI 需要写代码、运行脚本或操作文件时，它不会在你的本地机器上操作，而是连接到 E2B 的云端环境中进行。

核心特性

1. 安全性与隔离性 (Security & Isolation) ：
   • AI 生成的代码（可能包含错误或恶意逻辑）完全运行在云端沙箱中， 绝不会破坏你本地的电脑环境 。
   • 演示代码中，Agent 即使执行了 rm -rf / ，也只是删除了云端临时的沙箱，对宿主机毫发无损。
2. 持久化会话 (Long-running Sessions) ：
   • 沙箱可以保持运行状态。Agent 可以先创建一个文件（如演示中的 /home/user/hello.py ），然后在后续步骤中运行它。环境状态在会话期间是保持的。
3. 标准 Linux 环境 ：
   • 它提供标准的 Linux Shell。演示中 Agent 执行了 uname -a 和 python --version ，就像在真实的服务器上一样。

# 定义 E2B 沙箱后端类
# 该类继承自 BaseSandbox，实现了在 E2B 沙箱中执行命令的功能

import base64
from typing import Any, Optional

# 导入 DeepAgents 后端协议，定义了沙箱后端的接口
from deepagents.backends.protocol import (
    ExecuteResponse,
    FileDownloadResponse,
    FileUploadResponse,
    SandboxBackendProtocol,
)
# 导入基础沙箱类，用于实现沙箱后端的基本功能
from deepagents.backends.sandbox import BaseSandbox

try:
    from e2b import Sandbox
except ImportError:
    Sandbox = None

class E2BBackend(BaseSandbox):
    """E2B 沙箱后端实现，用于 DeepAgents。
    该后端使用 E2B（https://e2b.dev）提供安全、隔离的执行环境。
    """
    def __init__(
        self,
        template: str = "base",
        api_key: Optional[str] = None,
        timeout: Optional[int] = None,
        metadata: Optional[dict[str, str]] = None,
    ) -> None:
        """初始化 E2B 沙箱。

        参数:
            template: E2B 沙箱模板 ID（默认："base"）
            api_key: E2B API 密钥（可选，默认使用 E2B_API_KEY 环境变量）
            timeout: 沙箱超时时间（秒）
            metadata: 自定义沙箱元数据
        """
        if Sandbox is None:
            raise ImportError(
                "e2b package is not installed. "
                "Please install it with `pip install e2b`."
            )

        self.sandbox = Sandbox.create(
            template=template,
            api_key=api_key,
            timeout=timeout,
            metadata=metadata,
        )

    @property
    def id(self) -> str:
        """Unique identifier for the sandbox backend."""
        return self.sandbox.sandbox_id

    def execute(self, command: str) -> ExecuteResponse:
        """Execute a command in the sandbox."""
        try:
            # E2B commands.run returns CommandResult with stdout, stderr, exit_code
            result = self.sandbox.commands.run(command)

            # 返回执行结果，包含 stdout 是标准输出，stderr 是标准错误输出，exit_code 是退出码
            return ExecuteResponse(
                output=result.stdout + result.stderr,
                exit_code=result.exit_code,
                truncated=False,
            )
        except Exception as e:
            return ExecuteResponse(
                output=f"Error executing command: {str(e)}",
                exit_code=1,
                truncated=False,
            )

    def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
        """Upload multiple files to the sandbox."""
        responses = []
        for path, content in files:
            try:
                # Ensure directory exists before writing
                # We can use execute to mkdir -p
                parent_dir = path.rsplit("/", 1)[0]
                if parent_dir:
                    self.sandbox.commands.run(f"mkdir -p {parent_dir}")

                # Write file
                self.sandbox.files.write(path, content)
                responses.append(FileUploadResponse(path=path, error=None))
            except Exception as e:
                error_msg = str(e).lower()
                error = "invalid_path"
                if "permission" in error_msg:
                    error = "permission_denied"

                responses.append(FileUploadResponse(path=path, error=error))
        return responses

    def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
        """Download multiple files from the sandbox."""
        responses = []
        for path in paths:
            try:
                content = self.sandbox.files.read(path)
                # Ensure content is bytes
                if isinstance(content, str):
                    content = content.encode("utf-8")

                responses.append(FileDownloadResponse(path=path, content=content, error=None))
            except Exception as e:
                error_msg = str(e).lower()
                error = "invalid_path"
                if "not found" in error_msg:
                    error = "file_not_found"
                elif "directory" in error_msg:
                    error = "is_directory"
                elif "permission" in error_msg:
                    error = "permission_denied"

                responses.append(FileDownloadResponse(path=path, content=None, error=error))
        return responses

    def close(self):
        """Close the sandbox session."""
        self.sandbox.kill()

将自定义好的沙箱后端类注册到 DeepAgents 中

import asyncio
import os
import sys
from dotenv import load_dotenv
from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_core.messages import BaseMessage, ToolMessage

# 假设 e2b_backend.py 在同一目录下
try:
    from e2b_backend import E2BBackend
except ImportError:
    # 尝试从当前路径导入
    sys.path.append(os.path.dirname(os.path.abspath(__file__)))
    from e2b_backend import E2BBackend

load_dotenv(override=True)

async def setup_mcp_tools():
    """连接 Context7 MCP 服务器并获取工具"""
    print("正在连接 Context7 MCP 服务器...")
    try:
        client = MultiServerMCPClient({
            "context7": {
                "transport": "stdio",
                "command": "npx",
                "args": ["-y", "@upstash/context7-mcp@latest"],
            }
        })
        # 获取工具
        tools = await client.get_tools()
        return client, tools
    except Exception as e:
        return None, []

async def run_e2b_demo():
    # 1. 检查 API Key
    if not os.getenv("E2B_API_KEY"):
        print("❌ 错误: 未找到 E2B_API_KEY 环境变量。")
        return

    # 2. 初始化 MCP (用于查询文档等辅助任务)
    mcp_client, mcp_tools = await setup_mcp_tools()

    # 3. 初始化 E2B Backend
    try:
        backend = E2BBackend(template="base")
    except Exception as e:
        return

    try:
        # 4. 创建 DeepAgent
        llm = ChatOpenAI(model="gpt-4o", temperature=0)

        agent = create_deep_agent(
            model=llm,
            tools=mcp_tools,     # 赋予 MCP 工具能力
            backend=backend,     # 赋予 E2B 沙箱能力
            system_prompt="""你是一个拥有云端沙箱环境的高级技术助手。
            你的任务是演示如何在沙箱中进行操作。
            请执行以下步骤：
            1. 使用 'execute_command' 运行 'uname -a' 和 'python --version' 来展示环境信息。
            2. 创建一个 Python 脚本 '/home/user/hello.py'，内容是打印 'Hello from E2B Sandbox!'。
            3. 运行这个 Python 脚本并显示输出。
            """
        )

5. DockerBackend 使用 Docker 容器

作用与优势

• 安全隔离 (Security & Isolation)
  • 作用 ：Agent 的所有操作（文件读写、代码执行、系统命令）都被限制在 Docker 容器内部。
  • 优势 ：即使 Agent 产生幻觉执行了恶意代码（如删除系统文件），也只会破坏容器， 您的 Mac 宿主机毫发无损 。演示代码中的 auto_remove=True 确保任务结束后容器自动销毁，不留痕迹。
• 环境一致性 (Reproducibility)
  • 作用 ：代码指定了镜像 image="python:3.11-slim" 。
  • 优势 ：无论您的电脑安装的是 Python 3.9 还是 3.12，Agent 永远在一个干净、标准的 Python 3.11 环境中运行。这解决了“在我的机器上能跑”的经典依赖问题。
• 生命周期管理 (Lifecycle Management)
  • 作用 ： DockerBackend 自动处理容器的 启动 -> 连接 -> 执行 -> 销毁 全过程。
  • 优势 ：开发者无需手动编写复杂的 Docker 命令，像使用本地对象一样简单地调用 backend.execute() 或 backend.write_file() 。

# 这个 Backend 会在本地启动一个 Docker 容器，并将会话隔离在容器内部。

# - 核心功能 ：
#   - 自动生命周期管理：初始化时启动容器，结束时自动销毁 (auto_remove=True)。
#   - 高效文件传输：使用 tar 流在宿主机和容器之间传输文件，支持批量操作。
#   - 资源限制：支持设置 CPU (cpu_quota) 和内存 (memory_limit) 限制，防止 Agent 耗尽本机资源。
#   - 网络控制：可选禁用网络 (network_disabled=True) 以增强安全性。

import io
import tarfile
import time
import uuid
from typing import Optional

# 加载 DeepAgents 后端协议
from deepagents.backends.protocol import (
    ExecuteResponse,
    FileDownloadResponse,
    FileUploadResponse,
    SandboxBackendProtocol,
)

# 加载 DeepAgents 基础沙箱类
from deepagents.backends.sandbox import BaseSandbox

try:
    import docker
    from docker.errors import NotFound, APIError
except ImportError:
    docker = None

class DockerBackend(BaseSandbox):
    """Docker 沙箱后端实现，用于 DeepAgents。

    该后端使用本地 Docker 守护进程提供隔离的执行环境。
    需要安装 `docker` Python 包，并确保 Docker 守护进程正在运行。
    """
    def __init__(
        self,
        image: str = "python:3.11-slim",
        auto_remove: bool = True,
        cpu_quota: int = 50000,  # 50% CPU
        memory_limit: str = "512m",
        network_disabled: bool = False,
        working_dir: str = "/workspace",
        volumes: dict[str, dict[str, str]] | None = None,
    ) -> None:
        """初始化 Docker 沙箱。

        参数:
            image: 使用的 Docker 镜像（默认："python:3.11-slim"）
            auto_remove: 是否在关闭时移除容器（默认：True）
            cpu_quota: CPU 配额，单位为微秒（默认：50000）
            memory_limit: 内存限制（默认："512m"）
            network_disabled: 是否禁用网络访问（默认：False）
            working_dir: 容器内的工作目录（默认："/workspace"）
            volumes: Docker 卷配置，例如 {'/宿主机路径': {'bind': '/容器路径', 'mode': 'rw'}}
        """
        if docker is None:
            raise ImportError(
                "docker package is not installed. "
                "Please install it with `pip install docker`."
            )
        # 初始化 Docker 客户端，from_env() 会自动从环境变量中读取 Docker 配置
        self.client = docker.from_env()
        self.image = image
        self.auto_remove = auto_remove
        self.working_dir = working_dir
        self.volumes = volumes or {}
        self._container = None

        # Start container
        try:
            # Ensure image exists
            try:
                self.client.images.get(image)
            except NotFound:
                print(f"Pulling image {image}...")
                self.client.images.pull(image)

            self._container = self.client.containers.run(
                image,
                command="tail -f /dev/null",  # Keep container running
                detach=True,
                tty=True,
                cpu_quota=cpu_quota,
                mem_limit=memory_limit,
                network_disabled=network_disabled,
                working_dir=working_dir,
                volumes=self.volumes,
            )

            # Ensure working directory exists
            self.execute(f"mkdir -p {working_dir}")

        except Exception as e:
            raise RuntimeError(f"Failed to start Docker container: {e}")

    @property
    def id(self) -> str:
        """Unique identifier for the sandbox backend."""
        return self._container.id if self._container else "unknown"

    def execute(self, command: str) -> ExecuteResponse:
        """Execute a command in the sandbox."""
        if not self._container:
            return ExecuteResponse(
                output="Container not running",
                exit_code=1,
                truncated=False
            )

        try:
            # Docker exec_run 返回 (exit_code, output)
            # output 是字节类型
            # 使用列表形式的 cmd 以避免 shell 转义问题
            exit_code, output = self._container.exec_run(
                cmd=["bash", "-c", command],
                workdir=self.working_dir,
                demux=False # Combine stdout and stderr
            )

            return ExecuteResponse(
                output=output.decode("utf-8", errors="replace"),
                exit_code=exit_code,
                truncated=False,
            )
        except Exception as e:
            return ExecuteResponse(
                output=f"Error executing command: {str(e)}",
                exit_code=1,
                truncated=False,
            )

    def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
        """Upload multiple files to the sandbox using tar archive."""
        if not self._container:
            return [FileUploadResponse(path=p, error="permission_denied") for p, _ in files]

        responses = []

        # Create a tar archive in memory
        tar_stream = io.BytesIO()
        with tarfile.open(fileobj=tar_stream, mode='w') as tar:
            for path, content in files:
                # Docker put_archive expects relative paths inside the tar to be relative to the destination
                # But here we want absolute paths to be respected.
                # Actually put_archive extracts to a directory.
                # To support absolute paths, we should probably upload to root /?
                # Or handle relative paths relative to working_dir.

                # Let's handle paths:
                # If path is absolute, we strip leading / and upload to root.
                # If path is relative, we upload to working_dir.

                # Simplification: We will create a tar with full structure and extract to /

                # Normalize path
                if path.startswith("/"):
                    arcname = path.lstrip("/")
                    dest_path = "/"
                else:
                    arcname = path
                    dest_path = self.working_dir

                info = tarfile.TarInfo(name=arcname)
                info.size = len(content)
                info.mtime = time.time()
                tar.addfile(info, io.BytesIO(content))

                responses.append(FileUploadResponse(path=path, error=None))

        tar_stream.seek(0)

        try:
            # We extract to / to support absolute paths in the tar
            # Note: This assumes all files in the batch can be extracted to the same root.
            # If mixed absolute/relative, this might be tricky.
            # For robustness, we might need to upload one by one if paths are mixed,
            # or group them.
            # Strategy: Always extract to / (root), and ensure arcnames are full paths (without leading /)

            self._container.put_archive(
                path="/",
                data=tar_stream
            )
        except Exception as e:
            # Mark all as failed if batch fails
            return [FileUploadResponse(path=p, error="permission_denied") for p, _ in files]

        return responses

    def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
        """Download multiple files from the sandbox."""
        if not self._container:
            return [FileDownloadResponse(path=p, error="permission_denied") for p in paths]

        responses = []
        for path in paths:
            try:
                # get_archive returns a tuple (generator, stat)
                bits, stat = self._container.get_archive(path)

                # Reconstruct tar from bits
                file_content = io.BytesIO()
                for chunk in bits:
                    file_content.write(chunk)
                file_content.seek(0)

                # Extract file from tar
                with tarfile.open(fileobj=file_content, mode='r') as tar:
                    # There should be only one file/dir
                    member = tar.next()
                    if member.isdir():
                        responses.append(FileDownloadResponse(path=path, error="is_directory"))
                        continue

                    f = tar.extractfile(member)
                    if f:
                        content = f.read()
                        responses.append(FileDownloadResponse(path=path, content=content, error=None))
                    else:
                        responses.append(FileDownloadResponse(path=path, error="file_not_found"))

            except NotFound:
                responses.append(FileDownloadResponse(path=path, error="file_not_found"))
            except Exception as e:
                error_msg = str(e).lower()
                error = "invalid_path"
                if "permission" in error_msg:
                    error = "permission_denied"

                responses.append(FileDownloadResponse(path=path, content=None, error=error))
        return responses

    def close(self):
        """Close the sandbox session."""
        if self._container:
            try:
                if self.auto_remove:
                    self._container.remove(force=True)
                else:
                    self._container.stop()
            except Exception:
                pass
            self._container = None

使用自定义的 DockerBackend 实现沙箱环境

import asyncio
import os
import sys
from dotenv import load_dotenv
from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage

# 尝试导入 DockerBackend
try:
    from docker_backend import DockerBackend
except ImportError:
    sys.path.append(os.path.dirname(os.path.abspath(__file__)))
    from docker_backend import DockerBackend

load_dotenv(override=True)

async def run_docker_demo():
    # 1. 检查 Docker 环境
    try:
        import docker
        docker.from_env().ping()
    except Exception as e:
        return

    # 2. 初始化 Docker Backend
    try:
        backend = DockerBackend(
            image="python:3.11-slim",
            auto_remove=True
        )
    except Exception as e:
        return

    try:
        # 3. 创建 Agent
        llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

        agent = create_deep_agent(
            model=llm,
            backend=backend,
            system_prompt="""你是一个运行在 Docker 容器中的 AI 助手。
            你的任务是演示环境隔离性。

            请执行以下步骤：
            1. 运行 'cat /etc/os-release' 查看容器操作系统。
            2. 运行 'python --version' 确认 Python 环境。
            3. 创建文件 '/workspace/hello_docker.py'，内容为打印 'Hello from Docker Container!'。
            4. 运行该脚本。
            """
        )

DockerBackend 和 E2BBackend 的区别

6. StoreBackend 使用数据库存储

import asyncio
import os
import uuid
import traceback
from dotenv import load_dotenv

# LangChain / LangGraph Imports
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_core.messages import BaseMessage, ToolMessage
from langgraph.store.postgres import PostgresStore
from langgraph.checkpoint.memory import MemorySaver
from psycopg_pool import ConnectionPool

# DeepAgents Imports
from deepagents import create_deep_agent
from deepagents.backends import StoreBackend

load_dotenv(override=True)

DB_URI = "postgresql://myuser:123456@localhost:5432/mydatabase"

async def setup_mcp_tools():
    """
    连接 Context7 MCP 服务器并获取工具。
    """
    try:
        client = MultiServerMCPClient({
            "context7": {
                "transport": "stdio",
                "command": "npx",
                "args": ["-y", "@upstash/context7-mcp@latest"],
            }
        })

        # 获取工具列表
        tools = await client.get_tools()

        return client, tools
    except Exception as e:
        return None, []

async def run_store_backend_demo():
    """
    运行 StoreBackend 演示：
    展示如何使用 PostgreSQL 作为 DeepAgents 的文件系统后端。
    """
    print_header()

    # Step 1. 初始化 MCP 工具 (Connect to Context7 MCP Server)
    mcp_client, mcp_tools = await setup_mcp_tools()

    # Step 2. 初始化 PostgreSQL 连接池和存储组件 (Init DB & Store)
    # 使用 ConnectionPool 管理数据库连接
    # StoreBackend 需要同步的 ConnectionPool (因为它在线程中运行同步操作)
    try:
        with ConnectionPool(conninfo=DB_URI, kwargs={"autocommit": True}) as pool:
            # 2.1 初始化 Checkpointer (用于保存会话状态 / 聊天记录)
            # 注意: 这里使用 MemorySaver 是为了避开 langgraph-checkpoint-postgres 在某些环境下的异步兼容性问题
            checkpointer = MemorySaver()

            # 2.2 初始化 Store (用于保存文件 / 长期记忆)
            # DeepAgents 的 StoreBackend 会将文件系统操作映射到这个 PostgresStore
            store = PostgresStore(pool)

            # 2.3 执行数据库迁移 (首次运行需要初始化表结构)
            # 这会在数据库中创建 'store' 表，用于存储 JSON 数据 (即我们的文件)
            with pool.connection() as conn:
                with conn.cursor() as cur:
                    for migration in store.MIGRATIONS:
                        cur.execute(migration)

            # Step 3. 创建基于 StoreBackend 的 Agent (Create Agent)
            # StoreBackend 将文件系统操作映射到 LangGraph 的 Store
            # 这意味着文件将持久化存储在 PostgreSQL 数据库中，跨重启可用
            llm = ChatOpenAI(model="gpt-4o", temperature=0)

            # 定义 Backend Factory
            # 这是一个工厂函数，用于在运行时(runtime)把 StoreBackend 实例化。
            # create_deep_agent 会在内部把 LangGraph 的 runtime 对象作为参数 rt 传进来，
            # 这样 StoreBackend 就能拿到 store / checkpointer 等资源，实现文件系统到 PostgreSQL 的映射。
            backend_factory = lambda rt: StoreBackend(rt)

            agent = create_deep_agent(
                model=llm,
                tools=mcp_tools,
                backend=backend_factory,
                store=store,              # 关键: 必须传入 store 实例
                checkpointer=checkpointer, # 关键: 传入 checkpointer 实例
                system_prompt="""你是一个高级技术助手。
                你的任务是使用 Context7 工具查询关于 'DeepAgents StoreBackend' 的用法。
                查询后，创建一个总结文件 '/knowledge/store_backend_notes.md'，并写入关键信息。
                由于你使用的是 StoreBackend，这个文件将直接存储在 PostgreSQL 数据库中。
                最后，请读取该文件以验证存储成功。"""
            )
            
            # Step 4. 执行任务 (Execute Task)
            thread_id = str(uuid.uuid4())
            config = {"configurable": {"thread_id": thread_id}}

            task = "请查询 StoreBackend 的用法，并将总结写入 /knowledge/store_backend_notes.md，最后读取它验证。"

• Backend vs Checkpointer vs Store 这三者的区别

7. CompositeBackend 使用混合模式

在单一后端模式下，Agent 所有的文件操作（读、写、列出目录）都只能去往同一个地方（要么全是本地磁盘，要么全是 Docker 容器内）。而 CompositeBackend 允许你根据 文件路径前缀 ，将请求分发给不同的后端。

核心优势

1. 性能与开销优化 (Performance) 这是最关键的技术优势。
   • DockerBackend 的局限 : 向 Docker 容器内读写文件（尤其是大文件）需要经过 Docker Daemon 的 API (如 put_archive / get_archive )，涉及网络通信和打包解包，开销较大。
   • CompositeBackend 的解法 : 对于数据文件（ /data ），直接通过 FilesystemBackend 进行本地 I/O 操作， 完全绕过了 Docker API ，读写速度是操作系统原生的速度。
2. 计算与存储分离 (Decoupling)
   • 计算是临时的 : processor.py 脚本可能只需要运行一次，运行环境（Python 依赖）可能很复杂且容易冲突。放在 Docker 里最合适。
   • 数据是永恒的 : raw_metrics.txt 和 health_report.txt 是业务资产。通过路由直接落盘到宿主机，即使 Docker 容器崩溃、被删除或重启， 数据毫发无损且立即可在宿主机访问
3. 给予 Agent "混合云" 的能力，Agent 可以像人类工程师一样工作：
   • "我在临时的沙箱里写代码测试（Docker）。"
   • "测试好了，我把结果保存到公司的共享网盘里（Filesystem / Mount）。"

import asyncio
import shutil
import os
import time
from pathlib import Path
from dotenv import load_dotenv

# DeepAgents 导入
from deepagents import create_deep_agent
from deepagents.backends.composite import CompositeBackend
from deepagents.backends.filesystem import FilesystemBackend
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_core.messages import BaseMessage, ToolMessage

# 导入 DockerBackend
try:
    from docker_backend import DockerBackend
except ImportError:
    try:
        from deepagents.backends.docker import DockerBackend
    except ImportError:
        DockerBackend = None

async def setup_mcp_tools():
    try:
        client = MultiServerMCPClient({
            "context7": {
                "transport": "stdio",
                "command": "npx",
                "args": ["-y", "@upstash/context7-mcp@latest"],
            }
        })
        tools = await client.get_tools()
        return client, tools
    except Exception as e:
        return None, []

async def run_composite_demo():
    load_dotenv(override=True)
    print_header()

    if DockerBackend is None:
        return

    # Step 1
    host_work_dir = Path("workspace/data_analysis_project").resolve()
    if host_work_dir.exists():
        shutil.rmtree(host_work_dir)
    host_work_dir.mkdir(parents=True, exist_ok=True)

    container_mount_path = "/data"
    docker_volumes = {
        str(host_work_dir): {'bind': container_mount_path, 'mode': 'rw'}
    }

    # Step 2
    fs_backend = FilesystemBackend(root_dir=host_work_dir, virtual_mode=True)

    docker_backend = DockerBackend(
        image="python:3.11-slim",
        auto_remove=True,
        volumes=docker_volumes
    )

    routes = {
        container_mount_path: fs_backend
    }
    backend = CompositeBackend(default=docker_backend, routes=routes)

    # Step 3
    mcp_client, mcp_tools = await setup_mcp_tools()

    system_prompt = f"""你是一名在混合环境中工作的高级数据工程师。

    环境地图:
    1. 执行层 (根目录 `/`):
       - 临时的 Docker 容器。
       - 用于创建脚本 (`.py`) 和运行命令。
       - 这里的文会在会话结束后消失。

    2. 存储层 (`{container_mount_path}`):
       - 从宿主机挂载的持久化存储。
       - 用于存放 输入 数据和 输出 报告。
       - 这里的文件会永久保存。

    你的任务:
    1. **摄入**: 创建一个文件 `{container_mount_path}/raw_metrics.txt`，内容为 "CPU: 45%, Mem: 60%"。
       (注意: 这使用了 'write_file' 工具，该工具通过路由直接写入宿主机文件系统)。

    2. **处理**: 创建一个 Python 脚本 `/processor.py` (在根目录)，该脚本:
       - 读取 `{container_mount_path}/raw_metrics.txt`。
       - 计算 "健康分数" (模拟一下即可)。
       - 将报告写入 `{container_mount_path}/health_report.txt`。
       - 打印 "Analysis Complete"。

    3. **执行**: 使用 `python /processor.py` 运行脚本。
       (注意: 这在 Docker 内部运行。Docker 因为卷挂载能看到这些文件)。

    4. **验证**: 读取 `{container_mount_path}/health_report.txt` 并显示它。
    """

    agent = create_deep_agent(
        model=ChatOpenAI(model="gpt-4o", temperature=0),
        tools=mcp_tools,
        backend=backend,
        system_prompt=system_prompt
    )

5.3. interrupt_on 参数

1. Human-in-the-loop(HITL) 人工干预

1. Human-in-the-loop (HITL)：对于 write_file 或 delegate_task 等关键操作，利用 LangGraph 的中断机制加入人工审批。
2. 异步运行：DeepAgent 的任务通常耗时较长，务必使用异步 Webhooks 接收结果。
3. 监控与调试：强烈建议结合 LangSmith 使用。由于 DeepAgent 内部有复杂的子 Agent 递归调用，使用 LangSmith 的 Tracing 功能是排查问题的有效手段。
4. 后端挂载：在生产环境中，建议将 VFS 挂载到云端存储（如 S3），以防止容器重启导致 Agent 的"记忆"丢失。
5. interrupt_on：这个参数其实是一个 HITL 的开关，就是把 HITL 的中间件插入到 DeepAgent 的执行流程中，当 DeepAgent 执行到需要人工审批的操作时，就会中断执行，等待人工审批。
   • 类型: dict[str, bool | InterruptOnConfig]
   • 作用: 映射“工具名称”到“中断配置”。
   • 示例: interrupt_on={"write_file": True} 表示当 Agent 试图调用 write_file 工具时，程序会暂停（Suspend），等待人工（Human）在 LangGraph 层面进行 Approve 、 Reject 或 Edit 操作后才能继续。

import os
import asyncio
from typing import Optional, Set
from langchain_openai import ChatOpenAI
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend
from langgraph.checkpoint.memory import InMemorySaver
from dotenv import load_dotenv
from langchain_tavily import TavilySearch
from langchain.agents.middleware.human_in_the_loop import (
    HITLResponse,
    ApproveDecision,
    EditDecision,
    RejectDecision
)
from langgraph.types import Command
from langchain_core.messages import BaseMessage, ToolMessage, AIMessage

load_dotenv(override=True)

async def run_interrupt_test():
    """
    示例 1: 基础中断功能 (封装版)
    在工具调用前中断，让用户确认是否继续执行
    """
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    search_tool = TavilySearch(max_results=2)

    # 创建 Agent，设置在 "tools" 节点中断
    agent = create_deep_agent(
        model=llm,
        tools=[search_tool],
        backend=FilesystemBackend(root_dir="./workspace", virtual_mode=True),
        checkpointer=InMemorySaver(),  # 必需！用于支持中断和恢复
        interrupt_on={"tavily_search": True},  # 在特定工具调用时中断
    )

    task = "搜索 'Python 异步编程' 的最新信息，并创建一个总结文件"

    # 为了避免之前的状态干扰，我们使用一个新的 thread_id
    config = {"configurable": {"thread_id": "demo_basic_refactored_v1"}}

    # 追踪已打印的消息数量，避免重复打印
    message_history_len = 0

    # --- 第一次执行 ---
    async for event in agent.astream({"messages": [("user", task)]}, config=config):
        if "messages" in event:
            current_messages = event["messages"]
            if len(current_messages) > message_history_len:
                # 打印新增的消息
                for i in range(message_history_len, len(current_messages)):
                    msg = current_messages[i]
                    if msg.type == "ai":
                        if hasattr(msg, 'tool_calls') and msg.tool_calls:
                            print(f"🔧 AI 决定调用工具: {msg.tool_calls[0]['name']}")
                            print(f"   参数: {msg.tool_calls[0]['args']}")
                        elif msg.content:
                            print(f"💬 AI: {msg.content}")
                    elif msg.type == "tool":
                        print(f"✅ 工具输出: {msg.content[:100]}..." if len(msg.content) > 100 else f"✅ 工具输出: {msg.content}")

                message_history_len = len(current_messages)

    # 检查是否中断
    # 使用 aget_state (async) 获取状态
    state = await agent.aget_state(config)
    print(f"\n⏸️  执行状态: {state.next}")

    if state.tasks:
        print(f"\n--- 🛑 执行已暂停 (HITL Middleware) ---")
        last_message = state.values["messages"][-1]

        if hasattr(last_message, "tool_calls") and last_message.tool_calls:
            tool_call = last_message.tool_calls[0]

            # === 人工介入 ===
            approval = input("\n[管理员]: 是否批准执行此操作? (y/n/e[编辑]): ")

            if approval.lower() == 'y':
                print("\n[系统]: 操作已批准，继续执行...")

                hitl_response = HITLResponse(
                    decisions=[ApproveDecision(type="approve")]
                )

                # === 恢复执行 ===
                # 使用 Command(resume=...)
                async for event in agent.astream(
                    Command(resume=hitl_response),
                    config=config,
                    stream_mode="values"
                ):
                    if "messages" in event:
                        current_messages = event["messages"]
                        if len(current_messages) > message_history_len:
                            for i in range(message_history_len, len(current_messages)):
                                msg = current_messages[i]

                                # 优化打印逻辑，清晰展示 AI 回复
                                if msg.type == "tool":
                                    print(f"\n[工具输出]:\n{msg.content[:300]}..." if len(msg.content) > 300 else f"\n[工具输出]:\n{msg.content}")
                                elif msg.type == "ai":
                                    if msg.content:
                                        print(f"\n[AI 回复]:\n{msg.content}\n")
                                    elif msg.tool_calls:
                                        print(f"\n🔧 AI 决定调用工具: {msg.tool_calls[0]['name']}")
                                        print(f"   参数: {msg.tool_calls[0]['args']}")

                            message_history_len = len(current_messages)

            else:
                print("\n[系统]: 操作被拒绝或您选择了其他选项 (本演示仅处理 'y')。")
    else:
        print("流程已完成，没有触发中断。")
        if state.values.get("messages"):
            last_msg = state.values["messages"][-1]
            if last_msg.type == "ai" and last_msg.content:
                print(f"\n[最终回复]: {last_msg.content}")

if __name__ == "__main__":
    try:
        # asyncio.run(run_interrupt_test())
        await run_interrupt_test()
    except KeyboardInterrupt:
        print("\n程序已停止")